package com.atguigu.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.GmallConfig;
import com.atguigu.utils.DimUtil;
import com.atguigu.utils.ThreadPoolUtil;
import javassist.Loader;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T, T> implements AsyncJoinFunction<T>{

    private Connection connection;
    private ThreadPoolExecutor threadPoolExecutor;
    private SimpleDateFormat sdf;
    private String tableName;

    public AsyncDimFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();

    }

    public SimpleDateFormat getSimpleDateFormat(){
        sdf = new SimpleDateFormat("yyyy-MM-dd");
        return sdf;
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    // 查询维度表信息
                    JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, getKey(input));

                    // 关联维度表信息
                    join(input, dimInfo);

                    // 将结果输出
                    resultFuture.complete(Collections.singleton(input));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("超时！！！！！");
    }
}
